package ma.zhiqiang.dataline.connector.sink;

import cn.hutool.core.util.ClassUtil;
import cn.hutool.json.JSONUtil;
import ma.zhiqiang.dataline.core.dto.ColumnMap;
import ma.zhiqiang.dataline.core.engine.Sink;
import ma.zhiqiang.dataline.core.dto.SinkRecord;
import ma.zhiqiang.dataline.core.format.WriterFormatCsv;
import ma.zhiqiang.dataline.core.format.WriterFormatJson;

import java.util.Properties;

public class ConsoleSink implements Sink {
    Properties properties;
    String format;

    public ConsoleSink(Properties properties) {
        this.properties = properties;
        this.format = properties.getProperty("format","text");
    }

    @Override
    public Boolean writer(SinkRecord sinkRecord) {
        String outline = "";
        if("csv".equals(format) && (sinkRecord.getData() instanceof ColumnMap)){
            outline = new WriterFormatCsv().format(sinkRecord);
        }else if("json".equals(format) && (sinkRecord.getData() instanceof ColumnMap)){
            outline = new WriterFormatJson().format(sinkRecord);
        }else if(sinkRecord.getData() instanceof String){  //"text".equals(format)
            outline = sinkRecord.getData().toString();
        }else if(sinkRecord.getData() instanceof ColumnMap){
            outline = new WriterFormatJson().format(sinkRecord);
        }
        System.out.println(outline);
        return Boolean.TRUE;
    }
}
